package cn.lj.payment.config.redis;

import com.fasterxml.jackson.core.type.TypeReference;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.convert.converter.Converter;
import org.apache.commons.lang3.StringUtils;
import org.springframework.core.serializer.support.SerializingConverter;
import org.springframework.dao.DataAccessException;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.core.Cursor;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.ScanOptions;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.data.redis.core.script.RedisScript;
import org.springframework.data.redis.serializer.RedisSerializer;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;

import java.util.*;
import java.util.concurrent.TimeUnit;

//common3 用于校验 字符串类型 的jar
@Service("RedisServiceImpl")
public class RedisServiceImpl implements RedisServie {

    @Autowired
    private RedisTemplate redisTemplate;

    @Override
    public void set(String key, String keyFiled, Object object) {
        redisTemplate.opsForValue().set(key+":"+keyFiled,object);
    }

    @Override
    public<K,V> void putAll(String key, Map<K,V> map) {
        redisTemplate.opsForHash().putAll(key,map);
    }

    @Override
    public <T> Map hgetpushAll(String key,Class<T> clazz) {
        final byte[] rawKey = serializeKey(key);
        return (Map) redisTemplate.execute(new RedisCallback<Map<String, T>>() {
            @Override
            public Map doInRedis(RedisConnection connection) {
                Cursor<Map.Entry<byte[], byte[]>> curosr = connection.hScan(rawKey, ScanOptions.NONE);
                Map.Entry<byte[], byte[]> entry;
                Map<String, T> map = new LinkedHashMap();
                while(curosr.hasNext()){
                    entry = curosr.next();
                    map.put(deserializeKey(entry.getKey()), deserializeValue(entry.getValue(),clazz));
                }
                return map;
            }
        }, true);
    }

    @Override
    public Set<String> keys(String pattern) {
        return redisTemplate.keys(pattern);
    }

    @Override
    public Long dbSize() {
        return (Long) redisTemplate.execute(new RedisCallback<Long>() {
            @Override
            public Long doInRedis(RedisConnection connection) throws DataAccessException {
                return connection.dbSize();
            }
        });

    }

    @Override
    public Boolean flushDB() {
        return (Boolean) redisTemplate.execute(new RedisCallback<Boolean>() {
            @Override
            public Boolean doInRedis(RedisConnection connection) throws DataAccessException {
                connection.flushDb();
                return true;
            }
        });
    }

    @Override
    public void delete(final String key) {
        redisTemplate.delete(key);
    }

    @Override
    public void delete(Collection<String> keys) {
        redisTemplate.delete(keys);
    }

    @Override
    public void set(final String key, final Object value, final Long timeOutSeconds) {
        redisTemplate.execute(new RedisCallback<Object>() {
            @Override
            public Object doInRedis(RedisConnection connection) {
                byte[] k = serializeKey(key);
                byte[] v = serializeDefaultValue(value);
                if (timeOutSeconds == 0L) {// 不设置超时
                    connection.set(k, v);
                } else {
                    connection.setEx(k, timeOutSeconds, v);// 设置超时时间 单位秒
                }
                return null;
            }
        });
    }

    @Override
    public Boolean getWaitLock(String lockKey, String lockValue, Long lockSeconds, Long timeOutSeconds) {
        if (StringUtils.isBlank(lockKey) || StringUtils.isBlank(lockValue) || lockSeconds == null
                || timeOutSeconds == null) {
            System.out.println("参数异常，参数不能为空！");
            return false;
        }
        if (lockSeconds < 0 || timeOutSeconds < 0 ) {
            System.out.println("参数异常，时间不能为负数！");
            return false;
        }
        //获取锁开始时间戳
        Long startTime = System.currentTimeMillis();
        //获取锁超时时间戳
        Long timeOutTime = startTime + timeOutSeconds * 1000;

        while (true) {
            //尝试上锁
            Boolean lockflg = tryLock(lockKey, lockValue, lockSeconds);
            //上锁成功
            if (lockflg) {
                return true;
            }
            //当前时间大于超时时间戳则返回上锁失败
            if (System.currentTimeMillis() > timeOutTime) {
                return false;
            }
        }
    }

    @Override
    public Boolean tryLock(String lockKey, String lockValue, Long lockSeconds) {
        if (StringUtils.isBlank(lockKey) || StringUtils.isBlank(lockValue)
                || lockSeconds == null) {
            System.out.println("参数异常，参数不能为空");
            return false;
        }
        if (lockSeconds < 0) {
            System.out.println("参数异常，时间不能为负数！");
            return false;
        }
        String script = " local rtn = redis.call('set',KEYS[1],ARGV[1],'NX','EX',ARGV[2]) "
                + " if rtn == false then return false end return true";
        RedisScript<Boolean> redisScript = new DefaultRedisScript<>(script, Boolean.class);
        return (Boolean) redisTemplate.execute(redisScript, Collections.singletonList(lockKey), lockValue, lockSeconds);
    }

    @Override
    public Boolean unLock(String lockKey, String lockValue) {
        String script = " local rtn = redis.call('GET', KEYS[1]) "
                + " if rtn == ARGV[1] then redis.call('del', KEYS[1]) return true end return false";
        RedisScript<Boolean> redisScript = new DefaultRedisScript<>(script, Boolean.class);
        return (Boolean) redisTemplate.execute(redisScript, Collections.singletonList(lockKey), lockValue);
    }

    @Override
    public Boolean tryLockBatch(Set<String> lockKeySet, String lockValue, long lockSeconds) {
        if (CollectionUtils.isEmpty(lockKeySet) || StringUtils.isBlank(lockValue) || lockSeconds < 1) {
            System.out.println("参数异常，参数不能为空");
            return false;
        }

        String variableStr = String
                .format("local keyInsertedTable = {} local lockFlag = true local rtnValue = nil  "
                                + "local lockValue = '%s' local lockSecond = %d ",
                        lockValue, lockSeconds) + getAllKeyTableVariable(lockKeySet);
        String callStr = " for key, lockKey in pairs(allKeyTable) "
                + "do rtnValue = redis.call('set', lockKey, lockValue, 'NX', 'EX', lockSecond) "
                + "if rtnValue == false then lockFlag = false break else table.insert(keyInsertedTable, lockKey) end end";
        String judgeStr = " if (not lockFlag) then for k,v in pairs(keyInsertedTable) "
                + "do local existValue = redis.call('get', v) if(existValue == lockValue) "
                + "then redis.call('del', v) end end return false end return true";
        String LockScript = variableStr + callStr + judgeStr;

        DefaultRedisScript<Boolean> redisScript = new DefaultRedisScript<>(LockScript, Boolean.class);
        return (Boolean) redisTemplate.execute(redisScript, Collections.singletonList(""));
    }

    @Override
    public Boolean unlockBatch(Set<String> lockKeySet, String lockValue) {
        if (CollectionUtils.isEmpty(lockKeySet) || StringUtils.isBlank(lockValue)) {
            System.out.println("参数异常，参数不能为空");
            return false;
        }

        String variableStr = String.format("local existValue = nil local lockValue = '%s' ", lockValue)
                + getAllKeyTableVariable(lockKeySet);
        String callStr = " for key, lockKey in pairs(allKeyTable) do existValue = redis.call('GET', lockKey) "
                + "if existValue == lockValue then redis.call('del', lockKey) end existValue = nil end return true";
        String unLockScript = variableStr + callStr;

        DefaultRedisScript<Boolean> redisScript = new DefaultRedisScript<>(unLockScript, Boolean.class);
        return (Boolean) redisTemplate.execute(redisScript, Collections.singletonList(""));
    }

    private String getAllKeyTableVariable(Set<String> lockKeySet) {
        StringBuffer keyTableBuff = new StringBuffer(" local allKeyTable = {");
        for (String str : lockKeySet) {
            keyTableBuff.append(",'").append(str).append("'");
        }
        return keyTableBuff.append("}").toString().replaceFirst(",", "");
    }

    @Override
    public Long ttl(final String key) {
        Long seconds = (Long) redisTemplate.execute(new RedisCallback<Long>() {
            @Override
            public Long doInRedis(RedisConnection connection) {
                byte[] keybyte = serializeKey(key);
                if (connection.exists(keybyte)) {
                    return connection.ttl(keybyte);
                } else {
                    return 0L;
                }
            }
        });
        return seconds;
    }

    @Override
    public void set(final String key, final Object value) {
        set(key, value, 0L);
    }

    @Override
    public <T> T get(final String key) {
        T obj = (T) redisTemplate.execute(new RedisCallback<T>() {
            @SuppressWarnings("unchecked")
            @Override
            public T doInRedis(RedisConnection connection) {
                byte[] keybyte = serializeKey(key);
                if (connection.exists(keybyte)) {
                    byte[] bs = connection.get(keybyte);
                    return (T) deserializeDefaultValue(bs);
                } else {
                    return null;
                }
            }
        });
        return obj;
    }

    @Override
    public void setObj(String key, Object value) {
        redisTemplate.opsForValue().set(key, value);
    }

    @Override
    public void setObj(String key, Object value, Long timeOutSeconds) {
        redisTemplate.opsForValue().set(key, value, timeOutSeconds, TimeUnit.SECONDS);
    }

    @SuppressWarnings("unchecked")
    @Override
    public <T> T getObj(String key) {
        return (T) redisTemplate.opsForValue().get(key);
    }

    @Override
    public <T> T getObj(final String key, final TypeReference<T> type) {
        T obj = (T) redisTemplate.execute(new RedisCallback<T>() {
            @Override
            public T doInRedis(RedisConnection connection) {
                byte[] keybyte = serializeKey(key);
                if (connection.exists(keybyte)) {
                    byte[] bs = connection.get(keybyte);
                    return deserializeValue(bs,type);
                } else {
                    return null;
                }
            }
        });
        return obj;
    }

    @Override
    public <T> T getObj(final String key, final Class<T> clazz) {
        T obj = (T) redisTemplate.execute(new RedisCallback<T>() {
            @Override
            public T doInRedis(RedisConnection connection) {
                byte[] keybyte = serializeKey(key);
                if (connection.exists(keybyte)) {
                    byte[] bs = connection.get(keybyte);
                    return deserializeValue(bs,clazz);
                } else {
                    return null;
                }
            }
        });
        return obj;
    }

    @Override
    public <T> void addList(String key, List<T> list) {
        for(T t : list){
            redisTemplate.opsForList().rightPush(key, t);
        }
    }

    @Override
    public <T> void addList(String key, List<T> list, Long timeOutSeconds) {
        addList(key, list);
        redisTemplate.expire(key, timeOutSeconds, TimeUnit.SECONDS);
    }

    @Override
    public <T> void setList(String key, List<T> list) {
        delete(key);
        addList(key, list);
    }

    @Override
    public <T> void setList(String key, List<T> list, Long timeOutSeconds) {
        setList(key, list);
        redisTemplate.expire(key, timeOutSeconds, TimeUnit.SECONDS);
    }

    @SuppressWarnings("unchecked")
    @Override
    public <T> List<T> getList(String key) {
        return (List<T>) redisTemplate.opsForList().range(key, 0, -1);
    }

    @Override
    public <T> List<T> getList(String key,final Class<T> clazz) {
        final byte[] rawKey = serializeKey(key);
        return (List<T>) redisTemplate.execute(new RedisCallback<List<T>>() {
            @Override
            public List<T> doInRedis(RedisConnection connection) {
                List<byte[]> list = connection.lRange(rawKey, 0, -1);
                List<T> resultList = new ArrayList<>(list.size());
                for(byte[] bt : list){
                    resultList.add(deserializeValue(bt,clazz));
                }
                return resultList;
            }
        }, true);
    }

    @Override
    public <T> List<T> getList(String key, final TypeReference<T> type) {
        final byte[] rawKey = serializeKey(key);
        return (List<T>) redisTemplate.execute(new RedisCallback<List<T>>() {
            @Override
            public List<T> doInRedis(RedisConnection connection) {
                List<byte[]> list = connection.lRange(rawKey, 0, -1);
                List<T> resultList = new ArrayList<>(list.size());
                for(byte[] bt : list){
                    resultList.add(deserializeValue(bt,type));
                }
                return resultList;
            }
        }, true);

    }

    @Override
    public <T> void putHash(String key, String hashKey, T value) {
        redisTemplate.opsForHash().put(key, hashKey, value);
    }

    @Override
    public <T> void putHash(String key, String hashKey, T value, Long timeOutSeconds) {
        putHash(key, hashKey, value);
        redisTemplate.expire(key, timeOutSeconds, TimeUnit.SECONDS);

    }

    @Override
    public <K, V> void putHashAll(String key, Map<K, V> map) {
        redisTemplate.opsForHash().putAll(key, map);
    }

    @Override
    public <K, V> void putHashAll(String key, Map<K, V> map, Long timeOutSeconds) {
        putHashAll(key, map);
        redisTemplate.expire(key, timeOutSeconds, TimeUnit.SECONDS);
    }

    @SuppressWarnings("unchecked")
    @Override
    public <T> T getHash(String key, String hashKey) {
        return (T) redisTemplate.opsForHash().get(key, hashKey);
    }

    @Override
    public <T> T getHash(String key, String hashKey, final Class<T> clazz) {
        Boolean flg = redisTemplate.opsForHash().hasKey(key, hashKey);
        if(!flg){
            return null;
        }
        final byte[] rawKey = serializeKey(key);
        final byte[] rawHashKey = serializeKey(hashKey);
        T objT = (T) redisTemplate.execute(new RedisCallback<T>() {
            @Override
            public T doInRedis(RedisConnection connection) {
                byte[] rawHashValue =  connection.hGet(rawKey, rawHashKey);
                return deserializeValue(rawHashValue,clazz);
            }
        }, true);
        return objT;
    }

    @Override
    public <T> T getHash(String key, String hashKey, final TypeReference<T> type) {
        Boolean flg = redisTemplate.opsForHash().hasKey(key, hashKey);
        if(!flg){
            return null;
        }
        final byte[] rawKey = serializeKey(key);
        final byte[] rawHashKey = serializeKey(hashKey);
        T objT = (T) redisTemplate.execute(new RedisCallback<T>() {
            @Override
            public T doInRedis(RedisConnection connection) {
                byte[] rawHashValue =  connection.hGet(rawKey, rawHashKey);
                return deserializeValue(rawHashValue,type);
            }
        }, true);
        return objT;
    }

    @SuppressWarnings({ "unchecked", "rawtypes" })
    @Override
    public <T> List<T> getMHash(String key, List hashKeys) {
        return redisTemplate.opsForHash().multiGet(key, hashKeys);
    }

    @Override
    public <T> List<T> getMHash(String key, List<String> hashKeys, final Class<T> clazz) {
        final byte[] rawKey = serializeKey(key);
        final byte[][] rawHashKeys = new byte[hashKeys.size()][];
        for (int i = 0; i < hashKeys.size(); i++) {
            rawHashKeys[i] = serializeKey(hashKeys.get(i));
        }
        List<T> list = (List<T>) redisTemplate.execute(new RedisCallback<List<T>>() {
            @Override
            public List<T> doInRedis(RedisConnection connection) {
                List<byte[]> rawHashValues =  connection.hMGet(rawKey, rawHashKeys);
                List<T> resultList = new ArrayList<>();
                for(byte[] rawHashValue : rawHashValues){
                    resultList.add(deserializeValue(rawHashValue,clazz));
                }
                return resultList;
            }
        }, true);
        return list;
    }

    @Override
    public <T> List<T> getMHash(String key, List<String> hashKeys, final TypeReference<T> type) {
        final byte[] rawKey = serializeKey(key);
        final byte[][] rawHashKeys = new byte[hashKeys.size()][];
        for (int i = 0; i < hashKeys.size(); i++) {
            rawHashKeys[i] = serializeKey(hashKeys.get(i));
        }
        List<T> list = (List<T>) redisTemplate.execute(new RedisCallback<List<T>>() {
            @Override
            public List<T> doInRedis(RedisConnection connection) {
                List<byte[]> rawHashValues =  connection.hMGet(rawKey, rawHashKeys);
                List<T> resultList = new ArrayList<>();
                for(byte[] rawHashValue : rawHashValues){
                    resultList.add(deserializeValue(rawHashValue,type));
                }
                return resultList;
            }
        }, true);
        return list;
    }

    @SuppressWarnings("rawtypes")
    @Override
    public Map getHashAll(String key) {
        final byte[] rawKey = serializeKey(key);
        return (Map) redisTemplate.execute(new RedisCallback<Map>() {
            @SuppressWarnings("unchecked")
            @Override
            public Map doInRedis(RedisConnection connection) {
//                Long mapLen = connection.hLen(rawKey);
//                ScanOptions scanOptions = ScanOptions.scanOptions().count(mapLen).build();
                Cursor<Map.Entry<byte[], byte[]>> curosr = connection.hScan(rawKey,ScanOptions.NONE);
                Map.Entry<byte[], byte[]> entry;
                Map map = new LinkedHashMap();
                while(curosr.hasNext()){
                    entry = curosr.next();
                    map.put(deserializeKey(entry.getKey()), deserializeDefaultValue(entry.getValue()));
                }
                return map;
            }
        }, true);

    }

    @SuppressWarnings("rawtypes")
    @Override
    public <T> Map getHashAll(String key, final Class<T> clazz) {
        final byte[] rawKey = serializeKey(key);
        return (Map) redisTemplate.execute(new RedisCallback<Map<String, T>>() {
            @SuppressWarnings("unchecked")
            @Override
            public Map doInRedis(RedisConnection connection) {
                Cursor<Map.Entry<byte[], byte[]>> curosr = connection.hScan(rawKey,ScanOptions.NONE);
                Map.Entry<byte[], byte[]> entry;
                Map<String, T> map = new LinkedHashMap();
                while(curosr.hasNext()){
                    entry = curosr.next();
                    map.put(deserializeKey(entry.getKey()), deserializeValue(entry.getValue(),clazz));
                }
                return map;
            }
        }, true);
    }

    @Override
    public <V> Map<String,V> getHashAll(String key, final TypeReference<V> type) {
        final byte[] rawKey = serializeKey(key);
        return (Map<String, V>) redisTemplate.execute(new RedisCallback<Map<String,V>>() {
            @Override
            public Map<String,V> doInRedis(RedisConnection connection) {
//                Long mapLen = connection.hLen(rawKey);
//                ScanOptions scanOptions = ScanOptions.scanOptions().count(mapLen).build();
                Cursor<Map.Entry<byte[], byte[]>> curosr = connection.hScan(rawKey,ScanOptions.NONE);
                Map.Entry<byte[], byte[]> entry;
                Map<String,V> map = new LinkedHashMap<>();
                while(curosr.hasNext()){
                    entry = curosr.next();
                    map.put(deserializeKey(entry.getKey()), deserializeValue(entry.getValue(),type));
                }
                return map;
            }
        }, true);
    }

    @Override
    public void delHash(String key, Object... hashKeys) {
        redisTemplate.opsForHash().delete(key, hashKeys);
    }

    /** 序列号 Key，Value */
    public byte[] serializeKey(String key) {
        RedisSerializer<String> keySerializer = redisTemplate.getStringSerializer();
        if (keySerializer == null) {
            return key.getBytes();
        }
        return keySerializer.serialize(key);
    }

    String deserializeKey(byte[] value) {
        RedisSerializer<String> keySerializer = redisTemplate.getStringSerializer();
        if (keySerializer == null) {
            return value.toString();
        }
        return keySerializer.deserialize(value);
    }

    @SuppressWarnings({ "unchecked", "rawtypes" })
    public byte[] serializeDefaultValue(Object value) {
        RedisSerializer valueSerializer = redisTemplate.getDefaultSerializer();
        if (valueSerializer == null) {
            return String.valueOf(value).getBytes();
        }
        Converter<Object, byte[]> serializer = new SerializingConverter();
        byte[] v = null;
        try {
            v = valueSerializer.serialize(value);
        } catch (Exception ec) {
            v = serializer.convert(value);
        }
        return v;
    }

    @SuppressWarnings("rawtypes")
    public Object deserializeDefaultValue(byte[] value) {
        RedisSerializer valueSerializer = redisTemplate.getDefaultSerializer();
        if (valueSerializer == null) {
            return value.toString();
        }
        return valueSerializer.deserialize(value);
    }

    public <T> byte[] serializeValue(T value) {
        return JacksonUtil.toJsonBytes(value);
    }

    public <T> T deserializeValue(byte[] value, TypeReference<T> type) {
        return JacksonUtil.toObject(value, type);
    }

    @Override
    public Long increment(String key, int increment) {

        return redisTemplate.opsForValue().increment(key, increment);
    }

    @Override
    public void expire(String key, Long timeOutSeconds) {
        redisTemplate.expire(key, timeOutSeconds, TimeUnit.SECONDS);
    }

    public <T> T deserializeValue(byte[] value, Class<T> clazz) {
        return JacksonUtil.toJavaBean(value, clazz);
    }
}
